package net.wicp.tams.common.others.kafka;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

import org.apache.kafka.clients.producer.KafkaProducer;

import lombok.extern.slf4j.Slf4j;

/***
 * common.others.kafka.key.serializer=org.apache.kafka.common.serialization.StringSerializer
 * common.others.kafka.value.serializer=org.apache.kafka.common.serialization.ByteArraySerializer
 * 
 * @author Administrator
 *
 */
@Slf4j
public class KafkaProducerInst {
	private static KafkaProducerInst inst = new KafkaProducerInst();

	private Map<String, KafkaProducer<String, ?>> map = new HashMap<>();

	private KafkaProducerInst() {
	}

	public static KafkaProducerInst getInst() {
		return inst;
	}

	public <T> KafkaProducer<String, T> createObject(Class<?> classz) {
		String key = classz.getName();
		if (map.isEmpty() || !map.containsKey(key)) {
			Properties prop = KafkaTools.getProps(true);
			String valueprop = KafkaTools.getValueProp(classz, true);
			prop.put("value.serializer", valueprop);
			prop.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
			log.info("kafka producer 参数:");
			for (Object propele : prop.keySet()) {
				log.info("{}:{}", propele, prop.get(propele));
			}
			// link和batch不用加
			KafkaProducer<String, T> kafkaProducer = new KafkaProducer<>(prop);
			map.put(key, kafkaProducer);
			return kafkaProducer;
		} else {
			return (KafkaProducer<String, T>) map.get(key);
		}
	}

}
